package org.example.producer;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 *  分区策略测试2
 * @ClassName CallBackProducer
 * @Author 小坏
 * @Date 2021/10/29、19:50
 * @Version 1.0
 */
public class CallBackProducer {
    public static void main(String[] args) {
        Properties properties = new Properties();
        //指定连接的kafka集群
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);


        /**
         * 带回调函数的
         * 测试发送数据回调的是从零开始的
         */
//        for (int i = 0; i < 10; i++) {
//            producer.send(new ProducerRecord<>("aaa", "atguigu-" + i),
//                    (recordMetadata, e) -> {
//                        if (e == null) {
//                            System.out.println(recordMetadata.partition() + "--" + recordMetadata.offset());
//                        } else {
//                            e.getMessage();
//                        }
//
//                    });
//        }


/**
 * 使用倒数第三个api
 * 带分区和key的
 */
//        for (int i = 0; i < 10; i++) {
//            producer.send(new ProducerRecord<>("aaa", 0, "atguigu", "atguigu-" + i),
//                    (recordMetadata, e) -> {
//                        if (e == null) {
//                            System.out.println(recordMetadata.partition() + "--" + recordMetadata.offset());
//                        } else {
//                            e.getMessage();
//                        }
//
//                    });
//        }









/**
 * 方法的重载
 *  atguigu 哈希值 % 3
 *
 *  1--3
 * 1--4
 * 1--5
 * 1--6
 * 1--7
 * 1--8
 * 1--9
 * 1--10
 * 1--11
 * 1--12
 *
 * 都进入了一个分区
 *
 */
//        for (int i = 0; i < 10; i++) {
//            producer.send(new ProducerRecord<>("aaa", "atguigu", "atguigu-" + i),
//                    (recordMetadata, e) -> {
//                        if (e == null) {
//                            System.out.println(recordMetadata.partition() + "--" + recordMetadata.offset());
//                        } else {
//                            e.getMessage();
//                        }
//
//                    });
//        }


/**
 * 同步发送
 * 调用 .get(); 方法
 */
        for (int i = 0; i < 10; i++) {
            Future<RecordMetadata> send = producer.send(new ProducerRecord<>("aaa", "atguigu", "atguigu-" + i));
            try {
                send.get();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }


        producer.close();
    }
}
